package cn.edu.thu.laud.jdbc;


import static org.apache.cassandra.cql.jdbc.LaUDCassandraUtils.NO_RESULTSET;

import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLNonTransientException;
import java.sql.SQLWarning;
import java.sql.Statement;


import org.apache.cassandra.cql.jdbc.LaUDCassandraResultSet;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.hadoop.hive.service.HiveServerException;
import org.apache.thrift.TException;

import cn.edu.thu.laud.jdbc.util.LaUDJdbcUtils;
import cn.edu.thu.laud.server.thrift.LaUDResult;
import cn.edu.thu.laud.server.thrift.LaUDServerException;
public class LaUDStatement implements Statement {
	
	public static final ByteBuffer EMPTY_BYTEBUFFER = ByteBufferUtil.EMPTY_BYTE_BUFFER;
	private Statement  cassandraStatement;
	private Statement  hiveStatement;
	
	//private int statementType=LaUDJdbcUtils.TYPE_OTHER;
	
	private LaUDResult resultSet=null;

	private Compression compression=Compression.GZIP;
	
	private LaUDConnection connection;
	/**
	 * 是否进行转义
	 */
	boolean escapeProcessing=true;

	/**
	 * 
	 * @param laUDConnection
	 * @param cassandraStatement  CassandraStatement
	 * @param hiveStatement       HiveStatement
	 */
	public LaUDStatement(LaUDConnection laUDConnection,
			Statement cassandraStatement, Statement hiveStatement) {
		this.connection=laUDConnection;
		this.cassandraStatement=cassandraStatement;
		this.hiveStatement=hiveStatement;
	}


	@Override
	public boolean isWrapperFor(Class<?> iface) throws SQLException {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public <T> T unwrap(Class<T> iface) throws SQLException {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public void addBatch(String sql) throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public void cancel() throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public void clearBatch() throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public void clearWarnings() throws SQLException {
		hiveStatement.clearWarnings();
	}

	@Override
	public void close() throws SQLException {
		if(connection.includeHive)
			hiveStatement.close();
		cassandraStatement.close();
		connection.removeStatement(this);
        connection = null;
	}

	@Override
	public boolean execute(String sql) throws SQLException {
		try {
			resultSet=connection.execute(sql,compression);
		} catch (LaUDServerException e) {
			throw new SQLException(e);
		} catch (TException e) {
			throw new SQLException(e);
		} catch (InvalidRequestException e) {
			throw new SQLException(e);
		} catch (UnavailableException e) {
			throw new SQLException(e);
		} catch (TimedOutException e) {
			throw new SQLException(e);
		} catch (SchemaDisagreementException e) {
			throw new SQLException(e);
		} catch (HiveServerException e) {
			throw new SQLException(e);
		}
		if(resultSet.type==LaUDJdbcUtils.TYPE_CASSANDRA){
			this.connection.setCassandraResult(resultSet.cqlResult);
			return cassandraStatement.execute("");
		}else if(resultSet.type==LaUDJdbcUtils.TYPE_HIVE){
			if(connection.includeHive)
				return hiveStatement.execute("");
			else {
			 throw new SQLException("laud server do not run hive plungin.");
			}
		}else if(resultSet.type==LaUDJdbcUtils.TYPE_FILE){
			return true;
		}else if(resultSet.type==LaUDJdbcUtils.TYPE_MIX){
			this.connection.setCassandraResult(resultSet.cqlResult);
			cassandraStatement.execute("");
			return true;
		}
		return false;
	}

	@Override
	public boolean execute(String sql, int autoGeneratedKeys)
			throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public boolean execute(String sql, int[] columnIndexes) throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public boolean execute(String sql, String[] columnNames)
			throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public int[] executeBatch() throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public ResultSet executeQuery(String sql) throws SQLException {
		boolean haveResult=execute(sql);
		if(haveResult){
			return getResultSet();
		}else{
			throw new SQLNonTransientException(NO_RESULTSET);
		}
//		if(resultSet.type==LaUDJdbcUtils.TYPE_CASSANDRA){
//			if (resultSet.cqlResult == null)
//	            throw new SQLNonTransientException(NO_RESULTSET);
//	        return cassandraStatement.getResultSet();
//		}else if(resultSet.type==LaUDJdbcUtils.TYPE_HIVE){
//			return hiveStatement.getResultSet();
//		}else{
//			return getResultSet();
//		}
		//throw new SQLException("no resultset return.");
	}

	@Override
	@Deprecated
	public int executeUpdate(String sql) throws SQLException {
		execute(sql);
		if(resultSet.type==LaUDJdbcUtils.TYPE_CASSANDRA){
			return cassandraStatement.executeUpdate(sql);
		}else if(resultSet.type==LaUDJdbcUtils.TYPE_HIVE){
			return hiveStatement.executeUpdate(sql);
		}
		throw new SQLException("no supported.");
	}

	@Override
	public int executeUpdate(String sql, int autoGeneratedKeys)
			throws SQLException {
		throw new SQLException("no supported.");
	}

	@Override
	public int executeUpdate(String sql, int[] columnIndexes)
			throws SQLException {
		throw new SQLException("no supported.");
	}

	@Override
	public int executeUpdate(String sql, String[] columnNames)
			throws SQLException {
		throw new SQLException("no supported.");
	}

	@Override
	public Connection getConnection() throws SQLException {
		return this.connection;
	}
/**
 * return Cassandra Statement configure.
 */
	@Override
	public int getFetchDirection() throws SQLException {
		return cassandraStatement.getFetchDirection();
	}
/**
 * return hive statement configure
 */
	@Override
	public int getFetchSize() throws SQLException {
		return hiveStatement.getFetchSize();
	}
/**
 * return cassandra statement configure
 */
	@Override
	public ResultSet getGeneratedKeys() throws SQLException {
		return cassandraStatement.getGeneratedKeys();
	}
	/**
	 * return cassandra statement configure
	 */
	@Override
	public int getMaxFieldSize() throws SQLException {
		return cassandraStatement.getMaxFieldSize();
	}
	/**
	 * return hive statement configure
	 */
	@Override
	public int getMaxRows() throws SQLException {
		return hiveStatement.getMaxRows();
	}

	@Override
	public boolean getMoreResults() throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public boolean getMoreResults(int current) throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public int getQueryTimeout() throws SQLException {
		throw new SQLException("not support.");
	}

	@Override
	public ResultSet getResultSet() throws SQLException {
		if(resultSet==null)
			throw new SQLException("no resultset return.");
		if(resultSet.type==LaUDJdbcUtils.TYPE_CASSANDRA){
			if (resultSet.cqlResult == null)
	            throw new SQLNonTransientException(NO_RESULTSET);
	        return new LaUDCassandraResultSet((cassandraStatement.getResultSet()));
		}else if(resultSet.type==LaUDJdbcUtils.TYPE_HIVE){
			if(connection.includeHive)
				return hiveStatement.getResultSet();
		}else if(resultSet.type==LaUDJdbcUtils.TYPE_FILE){
			return new LaUDFileResultSet(resultSet.fileInfos,this.connection);
		}else if(resultSet.type==LaUDJdbcUtils.TYPE_MIX){
			return new LaUDMixResultSet(new LaUDCassandraResultSet(cassandraStatement.getResultSet()),new LaUDFileResultSet(resultSet.fileInfos,this.connection));
		}
		throw new SQLException("no resultset return.");
	}
/**
 * get cassandra config
 */
	@Override
	public int getResultSetConcurrency() throws SQLException {
		return cassandraStatement.getResultSetConcurrency();
	}
	/**
	 * get cassandra config
	 */
	@Override
	public int getResultSetHoldability() throws SQLException {
		return cassandraStatement.getResultSetHoldability();
	}
	/**
	 * get cassandra config
	 */
	@Override
	public int getResultSetType() throws SQLException {
		return cassandraStatement.getResultSetType();
	}
	/**
	 * get cassandra config
	 */
	@Override
	public int getUpdateCount() throws SQLException {
		return cassandraStatement.getUpdateCount();
	}

	@Override
	public SQLWarning getWarnings() throws SQLException {
		return hiveStatement.getWarnings();
	}

	@Override
	public boolean isClosed() throws SQLException {
		return this.connection==null;
	}

	@Override
	public boolean isPoolable() throws SQLException {
		return false;
	}

	@Override
	public void setCursorName(String name) throws SQLException {
		throw new SQLException("not support");
		
	}

	@Override
	public void setEscapeProcessing(boolean enable) throws SQLException {
		this.escapeProcessing=enable;
	}

	@Override
	public void setFetchDirection(int direction) throws SQLException {
		throw new SQLException("not support");
		
	}
	/**
	 * return hive config
	 */
	@Override
	public void setFetchSize(int rows) throws SQLException {
		hiveStatement.setFetchSize(rows);
	}
	/**
	 * return cassandra config
	 */
	@Override
	public void setMaxFieldSize(int max) throws SQLException {
		cassandraStatement.setMaxFieldSize(max);
	}
/**
 * return hive config
 */
	@Override
	public void setMaxRows(int max) throws SQLException {
		hiveStatement.setMaxRows(max);
	}

	@Override
	public void setPoolable(boolean poolable) throws SQLException {
		throw new SQLException("not support");
		
	}

	@Override
	public void setQueryTimeout(int seconds) throws SQLException {
		throw new SQLException("not support");
	}

}
